package com.zisu.rabbitproducer.component;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.Map;
import java.util.Random;
import java.util.UUID;

@Component
public class RabbitSenderEQ2 {
    @Autowired
    private RabbitTemplate rabbitTemplate;

    @Autowired
    private ConfirmBackSender confirmBackSender;


    final RabbitTemplate.ReturnCallback returnCallback = new RabbitTemplate.ReturnCallback(){
        //rabbitTemplate.setReturnCallback 会调用这个接口
        //第二种publisher-returns模式，该模式会在消息没有被路由到queues时将消息返回
        @Override
        public void returnedMessage(org.springframework.amqp.core.Message message, int replyCode, String replyText, String exchange, String routingKey) {
            System.out.println("消息未能成功路由到指定queues");
            System.out.println("return--message:" + new String(message.getBody()) + ",replyCode:" + replyCode
                    + ",replyText:" + replyText + ",exchange:" + exchange + ",routingKey:" + routingKey);

        }
    };


    /**
     * 确认消息 的回调监听接口，，用于确认消息是否被broker所收到(也就是 是否投递到exchange（不能保证是否投递到queue）)。。
     * 当消息被broker 成功接收到 后 会调用这个方法
     * */
    final RabbitTemplate.ConfirmCallback confirmCallback = new RabbitTemplate.ConfirmCallback() {
        /**
         * 参数：
         *      CorrelationData  作为一个唯一标识（ 生产者 消费者 消息确认的唯一标识 ）
         *      b                broker是否落盘成功
         *      s                失败的一些异常信息会送
         * */
        @Override
        public void confirm(CorrelationData correlationData, boolean b, String s) {
            //如果RabbitMQ 因为自身内部错
            //误导致消息丢失，就会发送一条nack (Basic . Nack ）命令，生产者应用程序同样可以在回调
            //方法中处理该nack 命令。

            if(b){
                System.err.println("消息ACK :"+b+",correlationDataId"+correlationData+"落盘 成功");
            }else {
                System.out.println("落盘失败，原因："+s);
            }
        }
    };


    /**
     *  @Description 向rabbitmq中发送消息（topic模式）
     *  @Param
     *          message     具体的消息内容
     *          properties  额外的附加属性
     *  @Return void
     *  @Author caixiang
     *  @Date 2020/6/17 
     **/
    public void sendTopic(Object message, Map<String,Object> properties, String routingKey) throws Exception{
        MessageHeaders messageHeaders = new MessageHeaders(properties);
        //message的Object 传进来的可能就是 泛型。。createMessage会自动生成 deliveryTag
        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);

        //唯一标识
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());

        //设置确认 confirm
        rabbitTemplate.setConfirmCallback(confirmCallback);

        /**
         * 声明消息处理器  (这个消息处理器 是在 发给broker 之前执行的（就是对你要发送给mq 的消息 进行封装处理 和 上面的confirmCallback不同）。)
         * 这个对消息进行处理  可以设置一些参数   对消息进行一些定制化处理
         * 我们这里  来设置消息的编码  以及消息的过期时间
         * 因为在.net 以及其他版本过期时间不一致   这里的时间毫秒值 为字符串
         * */
        MessagePostProcessor mpp = new MessagePostProcessor(){
            //当发送消息到broker 成功 之后调用下面的方法。
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                System.err.println("posted todo:"+message);
                //message.getMessageProperties()  下 可以进行各种 SET 属性
                return message;
            }
        };
        //发送端 要可以注重routingkey。。。因为exchange 要依据这个routingkey 把这条消息转发给队列。
        rabbitTemplate.convertAndSend("rc_exchange",
                routingKey,
                msg,
                mpp,
                cd
        );
    }


    /**
     *  @Description 向rabbitmq中发送消息（direct模式）
     *  @Param
     *          message     具体的消息内容
     *          properties  额外的附加属性
     *  @Return void
     *  @Author caixiang
     *  @Date 2020/6/17
     **/
    public void sendDirect(Object message, Map<String,Object> properties,String routingKey) throws Exception{
        MessageHeaders messageHeaders = new MessageHeaders(properties);
        //message的Object 传进来的可能就是 泛型。。createMessage会自动生成 deliveryTag
        Message<?> msg = MessageBuilder.createMessage(message,messageHeaders);

        //唯一标识
        CorrelationData cd = new CorrelationData(UUID.randomUUID().toString());


        //设置确认 confirm(),投递到Exchange 成功的时候 会调用下面这个回调方法（如果队列设置了 durable ，那也能通过这个回调函数确认消息是否已经存入磁盘）
        rabbitTemplate.setConfirmCallback(confirmCallback);
        //设置确认 returnedMessage();  当未投递到queue的时候 会调用下面这个方法。
        //注意 ：如果要设置setReturnCallback 那么 必须配置spring.rabbitmq.publisher-returns=true,spring.rabbitmq.template.mandatory=true
        rabbitTemplate.setReturnCallback(returnCallback);

        /**
         * 声明消息处理器  (这个消息处理器 是在 发给broker 之前执行的（就是对你要发送给mq 的消息 进行封装处理 和 上面的confirmCallback不同）。)
         * 这个对消息进行处理  可以设置一些参数   对消息进行一些定制化处理
         * 我们这里  来设置消息的编码  以及消息的过期时间
         * 因为在.net 以及其他版本过期时间不一致   这里的时间毫秒值 为字符串
         * */
        MessagePostProcessor mpp = new MessagePostProcessor(){
            //当发送消息到broker 成功 之后调用下面的方法。
            @Override
            public org.springframework.amqp.core.Message postProcessMessage(org.springframework.amqp.core.Message message) throws AmqpException {
                System.err.println("posted todo:"+message);
                //message.getMessageProperties()  下 可以进行各种 SET 属性
                int i = new Random().nextInt(10);
                System.out.println("优先级："+i);
                message.getMessageProperties().setPriority(i);
                return message;
            }
        };


        //发送端 要可以注重routingkey。。。因为exchange 要依据这个routingkey 把这条消息转发给队列。
        rabbitTemplate.convertAndSend("es2_exchange",
                routingKey,
                msg,
                mpp,
                cd
        );
    }
}
